package com.hrt.bigdata.mr.topn;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

/**
 * 求每个月最高温度的前两天
 *
 * 解题思路：map端的排序可以降低reduce端的处理成本。
 *          1.可以用最简单的方式，可以 年-月，values为所有数据，然后去reduce端去筛选处理。
 *          不过这种处理数据传输成本高，且reduce处理成本大
 *          2.让数据在map端排序，让数据在map端排序，按照年月升序，温度降序。且在reduce端自定义分组，让年月相同的分到同一组。
 *          这样在reduce端处理的时候，就可以直接取出第一条且再取出day不想等的一条即可。效率最高
 *          注意reduce的大坑。
 */
public class MyTopNDriver {


    public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration(true);

        conf.set("mapreduce.framework.name","local");
        conf.set("mapreduce.app-submission.cross-platform","true");


        String[] other = new GenericOptionsParser(conf, args).getRemainingArgs();

        Job  job = Job.getInstance(conf);
        job.setJarByClass(MyTopNDriver.class);

        job.setJobName("TopN");
        // 集群跑的时候需要些jar路径
        job.setJar("C:\\Users\\Administrator\\IdeaProjects\\target\\hadoop-hdfs-1.0-0.1.jar");
        //客户端规划的时候讲join的右表cache到mapTask出现的节点上
        // 先把文件上传到hdfs上
        job.addCacheFile(new Path("/data/topn/dict/dict.txt").toUri());

        //初学者，关注的是client端的代码梳理：因为把这块写明白了，其实你也就真的知道这个作业的开发原理；

        TextInputFormat.addInputPath(job,new Path(other[0]));
        Path outPath = new Path(other[1]);

        if(outPath.getFileSystem(conf).exists(outPath))  outPath.getFileSystem(conf).delete(outPath,true);
        TextOutputFormat.setOutputPath(job,outPath);


        /**key 自定义key */
        //map
        job.setMapperClass(TMapper.class);
        job.setMapOutputKeyClass(TKey.class);
        job.setMapOutputValueClass(IntWritable.class);

        /**partitioner  按  年，月  分区  -》  分区(一个分区可能会有多个组) > 分组  按 年分区！！！！！！
          分区器潜台词：满足  相同的key获得相同的分区号就可以~！*/
        job.setPartitionerClass(TPartitioner.class);
        /**sortComparator  年，月，温度  且  温度倒序*/
        job.setSortComparatorClass(TSortComparator.class);
        /**combine*/
//       job.setCombinerClass();

        //reducetask
        /**groupingComparator*/
        job.setGroupingComparatorClass(TGroupingComparator.class);
        //reduce
        job.setReducerClass(TReducer.class);

        job.waitForCompletion(true);



    }


}
